package somebak;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.Collector;

public class PayedTopnPerBrand {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt/");
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 订单主表
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("doitedu")
                .port(3306)
                .scanNewlyAddedTableEnabled(true)
                .databaseList("realtimedw")
                .tableList("realtimedw.oms_order")
                .username("root")
                .password("root")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();

        DataStreamSource<String> order = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "order");

        SingleOutputStreamOperator<OrderBean> process = order.keyBy(new KeySelector<String, Long>() {
            @Override
            public Long getKey(String json) throws Exception {
                JSONObject jsonObject = JSON.parseObject(json);
                JSONObject before = jsonObject.getJSONObject("before");
                JSONObject after = jsonObject.getJSONObject("after");

                return before == null ? after.getLong("id") : before.getLong("id");
            }
        }).process(new KeyedProcessFunction<Long, String, OrderBean>() {
            @Override
            public void processElement(String json, KeyedProcessFunction<Long, String, OrderBean>.Context context, Collector<OrderBean> collector) throws Exception {

                JSONObject jsonObject = JSON.parseObject(json);
                String op = jsonObject.getString("op");

                // 下单即支付
                if ("c".equals(op) || "r".equals(op)) {

                    JSONObject after = jsonObject.getJSONObject("after");
                    Integer status2 = after.getInteger("status");
                    if (status2 == 1) {
                        collector.collect(JSON.parseObject(after.toJSONString(), OrderBean.class));
                    }
                } else if ("u".equals(op)) {
                    JSONObject before = jsonObject.getJSONObject("before");
                    Integer statusBefore = before.getInteger("status");

                    JSONObject after = jsonObject.getJSONObject("after");
                    Integer statusAfter = after.getInteger("status");

                    // 未支付-> 支付
                    if (statusBefore == 0 && statusAfter > 0 && statusAfter < 4) {
                        if (statusAfter == 1) {
                            collector.collect(JSON.parseObject(after.toJSONString(), OrderBean.class));
                        }
                    }

                }
            }
        });

        tenv.createTemporaryView("order_mysql",process);
//        tenv.executeSql("select * from order_mysql").print();


        // 订单详情表
        tenv.executeSql(
                "CREATE TABLE item_mysql (    " +
                        "      id       BIGINT,  " +
                        "      order_id BIGINT,  " +
                        "      product_id BIGINT,                 " +
                        "      product_brand STRING,              " +
                        "      real_amount decimal(10,2),       " +
                        "     PRIMARY KEY (id) NOT ENFORCED       " +
                        "     ) WITH (                            " +
                        "     'connector' = 'mysql-cdc',          " +
                        "     'hostname' = 'doitedu'   ,          " +
                        "     'port' = '3306'          ,          " +
                        "     'username' = 'root'      ,          " +
                        "     'password' = 'root'      ,          " +
                        "     'database-name' = 'realtimedw',     " +
                        "     'table-name' = 'oms_order_item'     " +
                        ")"
        );

        // 关联
        tenv.executeSql(
                " CREATE TEMPORARY VIEW joined AS  SELECT  " +
                        "    o.id as order_id,                " +
                        "    o.payment_time - 8*60*60*1000 as payment_time, " +
                        "    o.status,                        " +
                        "    i.product_id,                    " +
                        "    i.product_brand,                 " +
                        "    i.real_amount                    " +
                        " from order_mysql o                  " +
                        " join item_mysql i                   " +
                        " on o.id = i.order_id                 "
        );

//        tenv.executeSql("select * from joined ").print();

        Table joinedTable = tenv.from("joined");
        DataStream<Row> joinedCdc = tenv.toChangelogStream(joinedTable);
        SingleOutputStreamOperator<Row> appendOnly = joinedCdc.filter(row -> row.getKind() == RowKind.INSERT);
        tenv.createTemporaryView("tmp",appendOnly, Schema.newBuilder()
                        .column("order_id", DataTypes.BIGINT())
                        .column("payment_time", DataTypes.BIGINT())
                        .column("status", DataTypes.INT())
                        .column("product_id", DataTypes.BIGINT())
                        .column("product_brand", DataTypes.STRING())
                        .column("real_amount", DataTypes.DECIMAL(10,2))
                        .columnByExpression("rt","to_timestamp_ltz(payment_time,3)")
                        .watermark("rt","rt - interval '0' second")
                .build());

//        tenv.executeSql("select * from tmp").print();

        tenv.executeSql(
                " SELECT                                                                                               "+
                        "     window_start,                                                                                    "+
                        " 	window_end,                                                                                        "+
                        "     product_brand,                                                                                   "+
                        " 	product_id,                                                                                        "+
                        "     amt,                                                                                             "+
                        " 	rn                                                                                                 "+
                        " FROM (                                                                                               "+
                        "     SELECT                                                                                           "+
                        "         window_start,window_end,                                                                     "+
                        "         product_brand,product_id,                                                                    "+
                        "         amt,                                                                                         "+
                        "     row_number() over(partition by window_start,window_end,product_brand order by amt desc) as rn    "+
                        "     FROM (                                                                                           "+
                        "       SELECT                                                                                         "+
                        "         window_start,window_end,                                                                     "+
                        "         product_brand,product_id,                                                                    "+
                        "         sum(real_amount) as amt                                                                      "+
                        "       from table(                                                                                    "+
                        "       CUMULATE(table tmp , descriptor(rt) , interval '1' minute,interval '24' HOUR)                  "+
                        "       )                                                                                              "+
                        "       group by                                                                                       "+
                        "       window_start,window_end,                                                                       "+
                        "       product_brand,product_id                                                                       "+
                        "     ) o1                                                                                             "+
                        " ) o2                                                                                                 "+
                        " where rn <=2                                                                                         "
        ).print();


        env.execute();

    }

}
